package engine

import (
	"04_pachong/fetcher"
	"log"
)

type ConcurrentEngine struct {
	Scheduler Scheduler

	WorkerCount int

	ItemChan chan interface{}
}

type Scheduler interface {
	ReadyNotifier

	WorkChan() chan Request

	Run()

	Submit(Request)
}

type ReadyNotifier interface {
	WorkerReady(chan Request)
}

func (c *ConcurrentEngine) Run(seeds ...Request) {

	if len(seeds) <= 0 {

		return
	}

	out := make(chan ParserResult)

	c.Scheduler.Run()

	for i := 0; i < c.WorkerCount; i++ {

		go createWorker(c.Scheduler.WorkChan(), out, c.Scheduler)
	}

	for _, v := range seeds {

		c.Scheduler.Submit(v)
	}

	for true {

		result := <-out

		for _, v := range result.Items {

			go func() {

				c.ItemChan <- v

			}()
		}

		for _, req := range result.Requests {

			c.Scheduler.Submit(req)
		}

	}

}

func createWorker(in chan Request, out chan ParserResult, s ReadyNotifier) {


	for true {

		s.WorkerReady(in)

		req := <-in

		//执行任务,爬取数据
		result, err := worker(req)

		if err != nil {

			log.Println("createWorker err : ", createWorker)
			continue
		}

		out <- result

	}

}

func worker(req Request) (ParserResult, error) {

	bytes, e := fetcher.Fetcher(req.HttpRequest)

	if e != nil {

		log.Printf("fetcher err : %s \n", e)

		return ParserResult{}, nil
	}

	//解析
	parserResult := req.ParserFunc(bytes)

	return parserResult, nil

}
